---
title: 创建和运行任务
description: 通过一套经典的用法来展现创建并运行任务的最简及最佳方法。
---
任务 Task 是可以独立运行的代码片段。Executor 管理着一个低级的 Thread 的对象池（Java 中并发的最原始形态）。可以创建一个任务，然后将其传递给一个 Executor 来执行。
## Task 和 Executor
在 Java 的早期版本中，要使用多线程就需要直接创建 Thread 对象，甚至是实现它们的子类以创建自定义的特殊“任务——线程”对象。你需要手动调用构造器，并且自行启动线程。

创建所有这些线程的开销都非常大，所以现在并不推荐这些手动的方法。Java 5 专门新增了一个类来处理**线程池**。不再需要为每个不同的任务类型都创建一个新的 Thread 子类，而只需要将任务创建为一个单独的类型，然后传给某个 ExecutorService 来运行该任务。该 ExecutorService 会为你管理多线程，并且在线程在完成任务后不会丢弃，而是会回收它们。

```java
package pub.wkq.thread;

public class NapTask implements Runnable{
    final int id;

    public NapTask(int id) {
        this.id = id;
    }

    @Override
    public void run() {
        new Nap(0.1);
        System.out.println(this + " " + Thread.currentThread().getName());
    }

    @Override
    public String toString() {
        return "NapTask{" +
                "id=" + id +
                '}';
    }
}
```
```java
package pub.wkq.thread;

import java.util.concurrent.TimeUnit;

public class Nap {
    public Nap(double t) {//秒
        try {
            TimeUnit.MILLISECONDS.sleep((long) (1000 * t));
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
    public Nap(double t,String msg){
        this(t);
        System.out.println(msg);
    }
}
```
调用 TimeUtil.MILLISECONDS.sleep(),会获得“当前线程“，并让它按参数中传递的时长睡眠，线程将被挂起，但是底层的处理器不会停止，操作系统会切换到某些其它任务。操作系统的任务管理器会定时检查 sleep() 是否到时间了，如果时间到了，线程会被唤醒，并继续分配给处理器时间。

sleep() 会抛出 InterruptedException 异常，这是 Java 早期设计的产物，通过立刻跳出任务来终止它们。由于这很容易产生不稳定的状态，后续便不再推荐如此终止任务了。

```java
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;

/**
 * @author 王开琦
 */
public class SingleThreadException {
    public static void main(String[] args) {
        ExecutorService exec = Executors.newSingleThreadExecutor();
        IntStream.range(0, 10).mapToObj(NapTask::new).forEach(exec::execute);
        System.out.println("All task submitted");
        exec.shutdown();
        while (!exec.isTerminated()) {
            System.out.println(Thread.currentThread().getName() + "await termination");
            new Nap(0.1);
        }
    }
}
```

newSingleThreadExecutor() 是 Executors 的工厂方法，用于创建特定类型的 ExecutorService。

我们创建了 10 个 NapTask 并将它们提交到了 ExecutorService ，它们会自行启动。于此同时，main 方法会继续处理其他的事情。当调用 exec.shutdown() 时，告诉 ExecutorService 完成所有已提交的任务，但是不再接受任何新任务。不过此时那些任务仍在运行，所以我们在退出 main()前必须等待这些任务完成。这是通过检查 exec.isTerminated()
的结果来实现的，所有的任务都完成后，该方法会返回 true

如果仅仅调用 exec.shutdown() ，程序会在所有任务完成后立即结束。也就是说，while(!exec.isTerminated()) 并不是必需的：

一旦调用了 exec.shutdown() ，此后再想提交任务，就会抛出 RejectExecutionException 异常：

```java
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author 王开琦
 */
public class MoreTasksAfterShutdown {
    public static void main(String[] args) {
        ExecutorService exec = Executors.newSingleThreadExecutor();
        exec.execute(new NapTask(1));
        exec.shutdown();
        exec.execute(new NapTask(99));
    }
}
```

## 使用更多的线程
使用多线程的目的几乎总是使任务完成的更快一些。除了 SingleThreadExecutor 中，还可以使用 CachedThreadPool。
```java
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;

/**
 * @author 王开琦
 */
public class CachedThreadPool {
    public static void main(String[] args) {
        ExecutorService exec = Executors.newCachedThreadPool();
        IntStream.range(0, 10).mapToObj(NapTask::new).forEach(exec::execute);
        exec.shutdown();
    }
}
```
因为不再用同一个线程来按顺序运行所有任务，而是每个任务都能获得自己的线程，所以它们都是并行的了。

```java title='接下来看下面这个例子'
/**
 * @author 王开琦
 */
public class InterferingTask implements Runnable {

    final int id;
    private static Integer val = 0;

    public InterferingTask(int id) {
        this.id = id;
    }
    @Override
    public void run() {
        for (int i = 0; i < 100; i++) {
            val++;
            System.out.println(id + " " + Thread.currentThread().getName() + " " + val);
        }
    }
}
```
每个任务都会使 val 自增 100 次。

```java title='使用 CachedThreadPool 来执行任务'
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;

public class CachedThreadPool2 {
    public static void main(String[] args) {
        ExecutorService exec = Executors.newCachedThreadPool();
        IntStream.range(0, 10).mapToObj(InterferingTask::new).forEach(exec::execute);
        exec.shutdown();
    }
}

```
输出结果每次都不一样，所有的任务都试图对 val 进行写操作。我们任务这样的类是非线程安全的，而使用 SingleThreadExecutor 则每次都能得到一致的结果了。

这是 SingleThreadExecutor 最大的好处————由于同时只会运行一个任务，这些任务永远不会互相影响，因此保证了线程安全性。这样的现象称为**线程封闭**。
## 生成结果
由于 InterferingTask 是一个 Runnable 的实现，它并没有返回值，因此只能通过副作用生成结果，即通过控制环境来生成结果（而不是直接返回结果）。副作用是并发编程的主要问题之一。InterferingTask 中的 val 称为**可变共享状态**（mutable shared state），正是它带来了问题：多个任务同时修改同一个变量会导致所谓的竞态条件。

避免竞态条件的最好方法是避免使用可变共享状态。

对于 InterferingTask ,如果能消除副作用，并且只返回任务结果，要达到这一目的，我们需要创建一个 Callable，而不是 Runnable。

```java title='call() 完全独立地生成结果，独立于其它的 CountingTask，所以不存在可变共享状态。'
package pub.wkq.thread;

import java.util.concurrent.Callable;

/**
 * @author 王开琦
 */
public class CountingTask implements Callable<Integer> {
    final int id;

    public CountingTask(int id) {
        this.id = id;
    }

    @Override
    public Integer call() throws Exception {
        Integer val = 0;
        for (int i = 0; i < 100; i++) {
            val++;
        }
        System.out.println(id + " " + Thread.currentThread().getName() + " " + val);
        return val;
    }
}
```

```java title='ExecutorService 可以通过调用 invokerAll() 来启动所有的 Callable：'
package pub.wkq.thread;

import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.IntStream;

/**
 * @author 王开琦
 */
public class CacheThread3 {
    public static Integer extractResult(Future<Integer> f) {
        try {
            return f.get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ExecutorService exec = Executors.newCachedThreadPool();
        List<CountingTask> tasks = IntStream.range(0, 10).mapToObj(CountingTask::new).toList();
        List<Future<Integer>> futures = exec.invokeAll(tasks);
        var sum = futures.stream().map(CacheThread3::extractResult).reduce(0, Integer::sum);
        System.out.println("sum = " + sum);
        exec.shutdown();
    }
}
```

只有当所有的任务都完成时，invokeAll() 才会返回由 Future 组成的 List ，每个 Future 对应一个任务。Future 是 Java5 引入的机制，它允许你提交一个任务，并且不需要等待它完成。

```java title='使用 ExecutorService.submit()'
package pub.wkq.thread;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * @author 王开琦
 */
public class Futures {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService exec = Executors.newSingleThreadExecutor();
        Future<Integer> future = exec.submit(new CountingTask(99));
        System.out.println("future.get() = " + future.get());
        exec.shutdown();
    }
}
```
当对尚未完成任务的 Future 调用 get() 方法时，调用会持续等待（阻塞），直到该 Future 返回结果可用。

由于 在调用 get() 时 Future 会阻塞，因此它只是将任务完成的时间推迟了。Future 是一个无效的解决办法，并不推荐使用它，更推荐的是 Java8 中的 CompletableFuture。

```java title='也可以使用更加优雅的方式——并行 Stream 来解决这个问题：'
package pub.wkq.thread;

import java.util.stream.IntStream;

/**
 * 0 ForkJoinPool.commonPool-worker-9 100
 * 8 ForkJoinPool.commonPool-worker-4 100
 * 5 ForkJoinPool.commonPool-worker-8 100
 * 4 ForkJoinPool.commonPool-worker-3 100
 * 9 ForkJoinPool.commonPool-worker-7 100
 * 2 ForkJoinPool.commonPool-worker-1 100
 * 3 ForkJoinPool.commonPool-worker-6 100
 * 6 main 100
 * 1 ForkJoinPool.commonPool-worker-2 100
 * 7 ForkJoinPool.commonPool-worker-5 100
 * 1000
 * @author 王开琦
 */
public class CountingStream {
    public static void main(String[] args) {
        System.out.println(IntStream.range(0, 10)
                .parallel()
                .mapToObj(CountingTask::new)
                .map(countingTask -> {
                    try {
                        return countingTask.call();
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                })
                .reduce(0, Integer::sum)
        );
    }
}
```

## 作为任务的 lambda 和方法引用
有了 lambda 和方法引用，将不再仅限于使用 Runnable 和 Callable。由于 Java8 通过匹配签名的方式支持 lambda 和 方法引用 ,也就是支持结构一致性，可以将非 Runnable 或 Callable 类型的参数传递给 ExecutorService。
```java
package pub.wkq.thread;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Lambda1
 * Not Runnable
 * Lambda2
 * Not Callable
 * @author 王开琦
 */
public class LambdasAndMethodReferences {
    public static void main(String[] args) {
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.submit(() -> System.out.println("Lambda1"));
        exec.submit(new NotRunnable()::go);
        exec.submit(()->{
            System.out.println("Lambda2");
            return 1;
        });
        exec.submit(new NotCallable()::get);
        exec.shutdown();

    }
}
class NotRunnable{
    public void go() {
        System.out.println("Not Runnable");
    }
}
class NotCallable{
    public Integer get() {
        System.out.println("Not Callable");
        return 1;
    }
}
```
此处，前两次的 submit() 调用可以替换为 execute() 调用。所有的 submit 调用都会返回 Future，可以在第二次的两个调用中获取结果。

